/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#include "log_recovery.h"

/**
 * @description: analyze阶段，需要获得脏页表（DPT）和未完成的事务列表（ATT）
 */
void RecoveryManager::analyze() {
	// 读入日志
	int offset_in_file = 0;
	lsn_t cpt_lsn = INVALID_LSN;
	while(true) {
		int size = disk_manager_->read_log(buffer_.get_data(), LOG_BUFFER_SIZE, offset_in_file);
		if (size <= 0) break;

		buffer_.set_size(size);
		// 读出日志记录
		int offset_in_buffer = 0;		
		while (offset_in_buffer < size) {
			// 如果剩余的日志不足以读出一个日志记录的头部，说明这个日志记录是不完整的
			if (uint32_t(size - offset_in_buffer) < LOG_HEADER_SIZE) {
				break;
			}
			std::shared_ptr<LogRecord> log_record = std::make_shared<LogRecord>();
			// 先读出log_record的头部
			log_record->deserialize(buffer_.get_data() + offset_in_buffer);
			// 如果剩余的日志不足以读出一个日志记录，说明这个日志记录是不完整的
			if (uint32_t(size - offset_in_buffer) < log_record->log_tot_len_) {
				break;
			}
			// 根据日志类型，读出具体的日志记录
			switch (log_record->log_type_) {
				case LogType::BEGIN:
					log_record = std::make_shared<BeginLogRecord>();
					break;
				case LogType::COMMIT:
					log_record = std::make_shared<CommitLogRecord>();
					break;
				case LogType::ABORT:
					log_record = std::make_shared<AbortLogRecord>();
					break;
				case LogType::UPDATE:
					log_record = std::make_shared<UpdateLogRecord>();
					break;
				case LogType::INSERT:
					log_record = std::make_shared<InsertLogRecord>();
					break;
				case LogType::DELETE:
					log_record = std::make_shared<DeleteLogRecord>();
					break;
				case LogType::END:
					log_record = std::make_shared<EndLogRecord>();
					break;
				case LogType::CHECKPOINT:
					cpt_lsn = log_record->lsn_;
					log_record = std::make_shared<CheckpointLogRecord>();
					break;
			}
			// 再读出具体的日志记录
			log_record->deserialize(buffer_.get_data() + offset_in_buffer);
			log_map_[log_record->lsn_] = log_record;
			offset_in_buffer += log_record->log_tot_len_;
			// 记录最后一个日志记录的lsn
			persist_lsn_ = std::max(persist_lsn_, log_record->lsn_);
		}
		offset_in_file += offset_in_buffer;
	}
	for (lsn_t log_lsn = cpt_lsn+1; log_lsn <= persist_lsn_; log_lsn++) {
		if (log_map_.find(log_lsn) == log_map_.end()) {
			continue;
		}
		std::shared_ptr<LogRecord> log_record = log_map_[log_lsn];
		switch (log_record->log_type_) {
			case LogType::BEGIN:
				att_[log_record->log_tid_] = log_record->lsn_;
				break;
			case LogType::COMMIT:
				att_.erase(log_record->log_tid_);
				break;
			case LogType::ABORT:
				att_[log_record->log_tid_] = log_record->lsn_;
				break;
			case LogType::UPDATE:
				att_[log_record->log_tid_] = std::max(att_[log_record->log_tid_], log_record->lsn_);
				break;
			case LogType::INSERT:
				att_[log_record->log_tid_] = std::max(att_[log_record->log_tid_], log_record->lsn_);
				break;
			case LogType::DELETE:
				att_[log_record->log_tid_] = std::max(att_[log_record->log_tid_], log_record->lsn_);
				break;
			case LogType::END:
				att_.erase(log_record->log_tid_);
				break;
			case LogType::CHECKPOINT:
				cpt_lsn = log_record->lsn_;
				break;
		}

		// 构造脏页表
		PageId page_id{INVALID_FD, INVALID_PAGE_ID};
		std::string table_name;
		if (log_record->log_type_ == LogType::INSERT) {
			std::shared_ptr<InsertLogRecord> insert_log_record = std::dynamic_pointer_cast<InsertLogRecord>(log_record);
			page_id.page_no = insert_log_record->rid_.page_no;
			table_name.assign(insert_log_record->table_name_, insert_log_record->table_name_size_);
		} else if (log_record->log_type_ == LogType::UPDATE) {
			std::shared_ptr<UpdateLogRecord> update_log_record = std::dynamic_pointer_cast<UpdateLogRecord>(log_record);
			page_id.page_no = update_log_record->rid_.page_no;
			table_name.assign(update_log_record->table_name_, update_log_record->table_name_size_);
		} else if (log_record->log_type_ == LogType::DELETE) {
			std::shared_ptr<DeleteLogRecord> delete_log_record = std::dynamic_pointer_cast<DeleteLogRecord>(log_record);
			page_id.page_no = delete_log_record->rid_.page_no;
			table_name.assign(delete_log_record->table_name_, delete_log_record->table_name_size_);
		}
		
		if (page_id.page_no != INVALID_PAGE_ID) {
			page_id.fd = sm_manager_->fhs_[table_name]->GetFd();
			if (dpt_.find(page_id) == dpt_.end()) {
				dpt_[page_id] = log_record->lsn_;
			}
		}
	}
}

/**
 * @description: 重做所有未落盘的操作
 */
void RecoveryManager::redo() {
	// 找到最小的 rec_lsn
	if (dpt_.empty()) {
		return;
	}
	lsn_t min_rec_lsn = dpt_.begin()->second;
	for (auto it = dpt_.begin(); it != dpt_.end(); it++) {
		min_rec_lsn = std::min(min_rec_lsn, it->second);
	}
	// 从最小的 rec_lsn 开始重做
	lsn_t lsn = min_rec_lsn;
	std::shared_ptr<LogRecord> log_record;
	while (log_map_.find(lsn) != log_map_.end()) {
		log_record = log_map_[lsn];
		switch (log_record->log_type_) {
			case LogType::INSERT: {
				std::shared_ptr<InsertLogRecord> insert_log_record = std::dynamic_pointer_cast<InsertLogRecord>(log_record);
				sm_manager_->redo_insert(insert_log_record->table_name_, insert_log_record->rid_, insert_log_record->insert_value_, insert_log_record->lsn_, context_);
				break;
			}
			case LogType::UPDATE: {
				std::shared_ptr<UpdateLogRecord> update_log_record = std::dynamic_pointer_cast<UpdateLogRecord>(log_record);
				sm_manager_->redo_update(update_log_record->table_name_, update_log_record->rid_, update_log_record->old_value_, update_log_record->new_value_, update_log_record->lsn_, context_);
				break;
			}
			case LogType::DELETE: {
				std::shared_ptr<DeleteLogRecord> delete_log_record = std::dynamic_pointer_cast<DeleteLogRecord>(log_record);
				sm_manager_->redo_delete(delete_log_record->table_name_, delete_log_record->rid_, delete_log_record->delete_value_, delete_log_record->lsn_, context_);
				break;
			}
			default:
				break;
		}
		lsn++;
	}
}

/**
 * @description: 回滚未完成的事务
 */
void RecoveryManager::undo() {
	lsn_t max_lsn;
	while (! att_.empty()) {
		max_lsn = INVALID_LSN;
		for (auto it = att_.begin(); it != att_.end(); it++) {
				max_lsn = std::max(max_lsn, it->second);
		}
		std::shared_ptr<LogRecord> log_record = log_map_[max_lsn];
		switch (log_record->log_type_) {
			case LogType::BEGIN: {
				att_.erase(log_record->log_tid_);
				continue;
			}
			case LogType::ABORT: {
				break;
			}
			case LogType::UPDATE: {
				std::shared_ptr<UpdateLogRecord> update_log_record = std::dynamic_pointer_cast<UpdateLogRecord>(log_record);
				sm_manager_->rollback_update(update_log_record->table_name_, update_log_record->rid_, update_log_record->old_value_, context_);
				break;
			}
			case LogType::INSERT: {
				std::shared_ptr<InsertLogRecord> insert_log_record = std::dynamic_pointer_cast<InsertLogRecord>(log_record);
				sm_manager_->rollback_insert(insert_log_record->table_name_, insert_log_record->rid_, context_);
				break;
			}
			case LogType::DELETE: {
				std::shared_ptr<DeleteLogRecord> delete_log_record = std::dynamic_pointer_cast<DeleteLogRecord>(log_record);
				sm_manager_->rollback_delete(delete_log_record->table_name_, delete_log_record->rid_, delete_log_record->delete_value_, context_);
				break;
			}
			default:
				assert(false);
		}
		// 更新att_
		if (log_record->prev_lsn_ == INVALID_LSN) {
			att_.erase(log_record->log_tid_);
		} else {
			att_[log_record->log_tid_] = log_record->prev_lsn_;
		}
	}
	// undo 完成后，将所有数据刷写到磁盘，并清空相关资源
	sm_manager_->flush_db();
	att_.clear();
	dpt_.clear();
	log_map_.clear();
}